package com.lhz.service.impl;

import cn.hutool.core.util.RandomUtil;
import cn.hutool.core.util.StrUtil;
import com.lhz.mapper.JobMapper;
import com.lhz.mapper.JobRegistryMapper;
import com.lhz.model.constant.RunnableConstant;
import com.lhz.model.entity.JobInfo;
import com.lhz.model.entity.JobRegistry;
import com.lhz.model.response.ResponseObject;
import com.lhz.service.ExecutorService;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.web.client.RestTemplate;

import java.time.LocalDateTime;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * @Author: LiHuaZhi
 * @Date: 2021/8/8 11:23
 * @Description:
 **/
@Service
@Slf4j
public class ExecutorServiceImpl implements ExecutorService {

    @Resource
    private JobMapper jobMapper;

    @Resource
    private RestTemplate restTemplate;

    @Resource
    private JobRegistryMapper jobRegistryMapper;

    @Override
    @Async
    public void startJob(JobInfo jobInfo) {

        String currentAddress = RunnableConstant.EXECUTOR_ADDRESS;

        // 其他机器地址
        Set<String> addressList = RunnableConstant.OTHER_EXECUTOR_ADDRESS;
        addressList.remove(currentAddress);
        for (String address : addressList) {
            String path = address + RunnableConstant.DISTRIBUTE_START_API;
            restTemplate.postForObject(path, jobInfo, ResponseObject.class);

        }
    }

    /**
     * 停止任务
     *
     * @param jobInfo
     */
    @Override
    @Async
    public void stopJob(JobInfo jobInfo) {
        String currentAddress = RunnableConstant.EXECUTOR_ADDRESS;

        // 其他机器地址
        Set<String> addressList = RunnableConstant.OTHER_EXECUTOR_ADDRESS;
        addressList.remove(currentAddress);
        for (String address : addressList) {
            String path = address + RunnableConstant.DISTRIBUTE_STOP_API;
            restTemplate.postForObject(path, jobInfo, ResponseObject.class);
        }
    }

    /**
     * 健康检测
     */
    @Override
    public void healthJob() {
        log.info("集群机器心跳检测中....");
        String currentAddress = RunnableConstant.EXECUTOR_ADDRESS;

        // 数据库中查询所有的机器注册地址
        List<JobRegistry> jobRegistries = jobRegistryMapper.listRegistry();
        if (CollectionUtils.isEmpty(jobRegistries)) {
            // 注册自身
            jobRegistry(currentAddress);
        }
        List<String> addressList = jobRegistries.stream().map(JobRegistry::getAddress).collect(Collectors.toList());
        if (!addressList.contains(currentAddress)) {
            // 注册自身
            jobRegistry(currentAddress);
        }

        addressList.remove(currentAddress);
        if (CollectionUtils.isEmpty(addressList)) {
            addressList.addAll(RunnableConstant.OTHER_EXECUTOR_ADDRESS);
        }
        for (String address : addressList) {
            try {
                String path = address + RunnableConstant.DISTRIBUTE_HEALTH_API;
                ResponseObject responseObject = restTemplate.postForObject(path, null, ResponseObject.class);
                if (!responseObject.getSuccess()) {
                    healthFail(address);
                }
                RunnableConstant.OTHER_EXECUTOR_ADDRESS.add(address);
                RunnableConstant.OTHER_EXECUTOR_FAIL_NUM.remove(address);
            } catch (Exception e) {
                // 心跳获取失败
                log.error("心跳获取失败", e);
                healthFail(address);
            } finally {
                Integer failNum = RunnableConstant.OTHER_EXECUTOR_FAIL_NUM.get(address);
                if (failNum != null && failNum > 3) {
                    RunnableConstant.OTHER_EXECUTOR_ADDRESS.remove(address);
                }
            }
        }
        log.info("集群机器心跳检测结束...");
    }

    private void healthFail(String address) {
        Integer failNum = RunnableConstant.OTHER_EXECUTOR_FAIL_NUM.get(address);
        failNum = failNum == null ? 1 : failNum + 1;
        // 记录心跳失败次数
        RunnableConstant.OTHER_EXECUTOR_FAIL_NUM.put(address, failNum);
    }

    private void jobRegistry(String currentAddress) {
        if (StrUtil.isNotBlank(currentAddress)) {
            // 查询自身是否在线
            JobRegistry jobRegistry = new JobRegistry();
            Long registryId = RandomUtil.randomLong();
            RunnableConstant.EXECUTOR_ID = registryId;
            jobRegistry.setId(registryId);
            jobRegistry.setAddress(currentAddress);
            jobRegistry.setRegistryTime(LocalDateTime.now());
            // 先更新
            int update = jobRegistryMapper.updateByAddress(jobRegistry);
            if (update <= 0) {
                jobRegistryMapper.insert(jobRegistry);
            }
        }
    }
}
